package HandshakeJoin;

import java.util.ArrayList;
import com.google.common.collect.Lists;
import org.apache.s4.base.Event;
import com.google.common.base.Preconditions;
import com.google.common.base.Joiner;

public class JoinResult extends Event
{
    transient private static final boolean removeJoinKeyField = true;

    private String joinId;
    private int joinKey;
    private ArrayList<Integer> rFields;
    private ArrayList<Integer> sFields;

    JoinResult(final String joinId, final Tuple r, final Tuple s) {
        final int rFieldIdx = r.getTargetFieldIdx();
        final int sFieldIdx = s.getTargetFieldIdx();
        Preconditions.checkState(r.getTargetField() == s.getTargetField());

        this.joinId = joinId;
        joinKey = r.getField(rFieldIdx);
        rFields = Lists.newArrayListWithCapacity(r.getSize() - 1);
        sFields = Lists.newArrayListWithCapacity(s.getSize() - 1);

        for (int i = 0; i < r.getFields().size(); ++i) {
            if (removeJoinKeyField && i == rFieldIdx) {
                continue;
            }

            rFields.add(r.getField(i));
        }

        for (int i = 0; i < s.getFields().size(); ++i) {
            if (removeJoinKeyField && i == sFieldIdx) {
                continue;
            }

            sFields.add(s.getField(i));
        }
    }

    public String getJoinId() {
        return joinId;
    }

    public int getJoinKey() {
        return joinKey;
    }

    public int getNumRFields() {
        return rFields.size();
    }

    public int getNumSFields() {
        return sFields.size();
    }

    public ArrayList<Integer> getRFields() {
        return rFields;
    }

    public ArrayList<Integer> getSFields() {
        return sFields;
    }

    public ArrayList<Integer> getRecord() {
        ArrayList<Integer> record = Lists.newArrayListWithCapacity(rFields
                .size() + sFields.size() + 1);

        record.add(joinKey);
        for (int value : rFields) {
            record.add(value);
        }
        for (int value : sFields) {
            record.add(value);
        }

        return record;
    }

    public String toString() {
        Joiner joinerComma = Joiner.on(", ").skipNulls();

        return "[" + joinerComma.join(getRecord()) + "]";
    }
}